package com.kerraway.config.sharding;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.Range;
import com.kerraway.config.SpringContextHolder;
import com.kerraway.dao.OrderRepository;
import com.kerraway.model.Order;
import com.kerraway.util.DateUtil;
import com.kerraway.util.JacksonUtil;
import io.shardingsphere.api.algorithm.sharding.ListShardingValue;
import io.shardingsphere.api.algorithm.sharding.PreciseShardingValue;
import io.shardingsphere.api.algorithm.sharding.RangeShardingValue;
import io.shardingsphere.api.algorithm.sharding.ShardingValue;
import io.shardingsphere.api.algorithm.sharding.complex.ComplexKeysShardingAlgorithm;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author 小柯
 * @date 2019/07/16
 */
@Slf4j
public class OrderLogShardingAlgorithm implements ComplexKeysShardingAlgorithm {

  private static final String ORDER_LOG_TABLE_LOGIC_NAME = "t_order_logs";
  private static final String ORDER_LOG_TABLE_PHYSICAL_NAME_FORMAT = "t_order_logs_%d_%s";
  private static final List<Integer> ORDER_LOG_TABLE_PHYSICAL_NAME_IDX_ONES = ImmutableList.of(0, 1);
  private static final String ORDER_LOG_COLUMN_NAME_ORDER_ID = "order_id";
  private static final String ORDER_LOG_COLUMN_NAME_CREATED_AT = "created_at";

  /**
   * 用于查看 OrderLogShardingAlgorithm 在什么时候被初始化
   */
  public OrderLogShardingAlgorithm() {
    logger.info("------------------init OrderLogShardingAlgorithm---------------------");
  }

  /**
   * Do sharding.
   * <p>
   * SQL 与 {@link ShardingValue} 实现类的对应关系：
   * insert 语句：{@link ListShardingValue}
   * = 和 in 查询语句：{@link ListShardingValue}
   * between 查询语句：{@link RangeShardingValue}
   * > 查询语句：不走分表逻辑？
   *
   * @param availableTargetNames available data sources or tables's names
   *                             这里传入的是配置 actual-data-nodes: all_tests.t_order_logs_0_201907
   * @param shardingValues       sharding values
   * @return sharding results for data sources or tables's names
   */
  @Override
  public Collection<String> doSharding(Collection<String> availableTargetNames, Collection<ShardingValue> shardingValues) {

    logger.info("--> availableTargetNames: {}", JacksonUtil.writeValue(availableTargetNames));
    logger.info("--> shardingValues: {}", JacksonUtil.writeValue(shardingValues));

    boolean isValid = !CollectionUtils.isEmpty(shardingValues) && shardingValues.stream()
        .allMatch(value -> ORDER_LOG_TABLE_LOGIC_NAME.equals(value.getLogicTableName()));
    Assert.isTrue(isValid, "shardingValues must be valid.");

    Map<String, Collection<Comparable<?>>> columnValuesMap = shardingValues.stream()
        .collect(Collectors.toMap(ShardingValue::getColumnName, this::getValues));
    Assert.isTrue(columnValuesMap.containsKey(ORDER_LOG_COLUMN_NAME_CREATED_AT),
        "shardingValues must contain " + ORDER_LOG_COLUMN_NAME_CREATED_AT + ".");

    List<Integer> tableIdxOnes = columnValuesMap.getOrDefault(ORDER_LOG_COLUMN_NAME_ORDER_ID, Collections.emptyList())
        .stream()
        .filter(Integer.class::isInstance)
        .map(Integer.class::cast)
        .map(orderId -> {

          //测试访问数据库
          testAccessDb(orderId);

          return orderId % 2;
        })
        .distinct()
        .collect(Collectors.toList());
    if (tableIdxOnes.isEmpty()) {
      tableIdxOnes = ORDER_LOG_TABLE_PHYSICAL_NAME_IDX_ONES;
    }
    List<String> tableIdxTwos = columnValuesMap.get(ORDER_LOG_COLUMN_NAME_CREATED_AT)
        .stream()
        .filter(Date.class::isInstance)
        .map(Date.class::cast)
        .map(date -> DateUtil.format(date, "yyyyMM"))
        .distinct()
        .collect(Collectors.toList());

    List<String> tableNames = new ArrayList<>(tableIdxOnes.size() * tableIdxTwos.size());
    for (Integer tableIdxOne : tableIdxOnes) {
      for (String tableIdxTwo : tableIdxTwos) {
        String tableName = String.format(ORDER_LOG_TABLE_PHYSICAL_NAME_FORMAT, tableIdxOne, tableIdxTwo);
        tableNames.add(tableName);
      }
    }

    logger.info("--> return tableNames: {}", tableNames);

    return tableNames;
  }

  private void testAccessDb(Integer orderId) {
    //测试查询数据库操作
    //Order order = OrderDaoFactory.orderDao().findFirstByOrderId(orderId);
    //logger.info("==> {}", order);

    OrderRepository orderDao = SpringContextHolder.getBean(OrderRepository.class);
    Order order1 = orderDao.findFirstByOrderId(orderId);
    logger.info("order 1: {}", order1);

    Order order2 = new Order();
    order2.setOrderId(new Random().nextInt(Integer.MAX_VALUE));
    order2.setUserId(new Random().nextInt(10) + 1);
    order2.setCreatedAt(new Date());
    order2.setUpdatedAt(new Date());
    orderDao.save(order2);
    logger.info("order 2: {}", order2);
  }

  /**
   * SQL 与 {@link ShardingValue} 实现类的对应关系：
   * insert 语句：{@link ListShardingValue}
   * = 和 in 查询语句：{@link ListShardingValue}
   * between 查询语句：{@link RangeShardingValue}
   * > 查询语句：不走分表逻辑？
   * <p>
   * 什么情况下，对应 {@link PreciseShardingValue}？
   *
   * @param shardingValue sharding value
   * @return values
   */
  private Collection<Comparable<?>> getValues(ShardingValue shardingValue) {
    Assert.notNull(shardingValue, "shardingValue must not be null.");

    if (shardingValue instanceof PreciseShardingValue) {
      return Collections.singletonList(((PreciseShardingValue) shardingValue).getValue());
    }
    if (shardingValue instanceof ListShardingValue) {
      return ((ListShardingValue<Comparable<?>>) shardingValue).getValues();
    }
    if (shardingValue instanceof RangeShardingValue) {
      Range<Comparable<?>> range = ((RangeShardingValue) shardingValue).getValueRange();
      return getValues(range);
    }
    throw new IllegalArgumentException("Unsupported sharding value type " + shardingValue.getClass() + ".");
  }

  /**
   * 只支持 {@link Integer} 和 {@link Date}
   *
   * @param range 范围
   * @return values
   */
  private Collection<Comparable<?>> getValues(Range<Comparable<?>> range) {
    Comparable<?> lowerEndpoint = range.lowerEndpoint();
    Comparable<?> upperEndpoint = range.upperEndpoint();
    if (lowerEndpoint instanceof Integer) {
      return IntStream.rangeClosed((Integer) lowerEndpoint, (Integer) upperEndpoint)
          .boxed()
          .collect(Collectors.toList());
    }
    if (lowerEndpoint instanceof Date) {
      Date cursor = (Date) lowerEndpoint;
      List<Comparable<?>> dates = new ArrayList<>(16);
      while (!cursor.after((Date) upperEndpoint)) {
        dates.add(cursor);
        cursor = DateUtils.addMonths(cursor, 1);
      }
      return dates;
    }
    throw new IllegalArgumentException("Unsupported range value type " + lowerEndpoint.getClass() + ".");
  }

}
